package com.cskaoyan.kpi.fsg;

import com.cskaoyan.kpi.Summary;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.util.*;

/**
 * @ClassName FSGReducer
 * @Description: TODO
 * @Author 远志 zhangsong@cskaoyan.onaliyun.com
 * @Date 2023/2/17 21:37
 * @Version V1.0
 **/
public class FSGReducer extends Reducer<Text, IntWritable, Summary, Text> {

    /**
     * 没有重写之前的TreeMap，不允许存储相同的key，会出现覆盖的情况
     * 如果两个商品的访问量刚刚好相同，那么便会出现覆盖的情况，造成失真
     */
    private TreeMap<Integer, String> fsg = new TreeMap<>(new Comparator<Integer>() {
        @Override
        public int compare(Integer o1, Integer o2) {
            if(o1.intValue() >= o2.intValue()){
                return 1;
            }
            return -1;
        }
    });



    @Override
    protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
        //key是2022-02-17:475 value是对应商品的次数 我们取前三
        //确保reducer只有一个，我们将键值对反过来存入treemap中，会自动排序
        //当数据大于3时，将最前面的去掉
        int sum = 0;
        for (IntWritable value : values) {
            sum += value.get();
        }
        fsg.put(sum, key.toString());
        if(fsg.size() > FrequentlySearchGoods.N){
            //删除第一组键值对
            fsg.pollFirstEntry();
        }
    }

    @Override
    protected void cleanup(Context context) throws IOException, InterruptedException {

        for (int i = 0; i < FrequentlySearchGoods.N; i++) {
            //循环N次，每次poll出最后一个，也就是最大值
            Map.Entry<Integer, String> integerStringEntry = fsg.pollLastEntry();
            Integer number = integerStringEntry.getKey();
            String text = integerStringEntry.getValue();
            String[] split = text.split(":");
            String date = split[0];
            String goodsId = split[1];
            Summary searchGoods = new Summary(null, date, "FrequentlySearchGoods", number + "", goodsId);
            context.write(searchGoods, new Text());
        }

    }
}
